package org.apm.data.check;

import java.util.HashSet;
import java.util.Set;

import org.apm.data.pusher.CommonMethod;
import org.elasticsearch.action.search.ClearScrollRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.filter.Filter;
import org.elasticsearch.search.aggregations.bucket.filter.FilterAggregationBuilder;
import org.joda.time.DateTime;

public class DataCorrectnessCheck {
    private Client client = CommonMethod.onlineClient();
    private String masterIp = "183.129.190.82";

    public DataCorrectnessCheck(String masterIp) {
        this.masterIp = masterIp;
    }

    public DataCorrectnessCheck(Client client, final String masterIp) {
        this.client = client;
    }

    private Set<String> getAllTransactionId(final DateTime day) {
        Set<String> transactionIds = new HashSet<>();

        String from = day.toString("yyyy-MM-dd") + " 00:00:00";
        String to = day.toString("yyyy-MM-dd") + " 23:59:59";

        QueryBuilder query = QueryBuilders.boolQuery()
                .filter(QueryBuilders.rangeQuery("DefaultMessage.startTime").from(from).to(to));

        Scroll scroll = new Scroll(new TimeValue(60000));
        SearchResponse scrollResp = client.prepareSearch("apm2.0-" + masterIp + "-javatracedata-1").setQuery(query)
                .addStoredField("DefaultMessage.transactionId").setScroll(scroll).setSize(1000).execute().actionGet();

        ClearScrollRequestBuilder clearRequest = client.prepareClearScroll();
        System.out.println("[Analyzer-After][" + masterIp + "] total: " + scrollResp.getHits().getTotalHits() + ".");
        while (true) {

            // Break condition: No hits are returned
            if (scrollResp.getHits().getHits().length == 0) {
                System.out.println("[Analyzer-After][" + masterIp + "] finished.");
                break;
            } else {
                // System.out.println("[Analyzer-After][" + masterIp + "] " +
                // scrollResp.getHits().getHits().length + ".");
            }

            for (SearchHit hit : scrollResp.getHits()) {
                String transactionId = hit.getFields().get("DefaultMessage.transactionId").getValue().toString();
                transactionIds.add(transactionId);
            }

            scrollResp = client.prepareSearchScroll(scrollResp.getScrollId()).setScroll(new TimeValue(60000)).execute()
                    .actionGet();
        }

        clearRequest.addScrollId(scrollResp.getScrollId()).execute().actionGet().isSucceeded();
        clearRequest.execute().actionGet();

        return transactionIds;
    }

    public boolean checkTransactionId(final String transactionId) {
        boolean correctness = true;

        QueryBuilder queryJava = QueryBuilders.boolQuery()
                .filter(QueryBuilders.termQuery("DefaultMessage.transactionId", transactionId));
        FilterAggregationBuilder aggJava = AggregationBuilders.filter("filter",QueryBuilders.termQuery("DefaultMessage.statusCode", 888));
        SearchResponse srJava = client.prepareSearch("apm2.0-" + masterIp + "-javatracedata-1").setQuery(queryJava)
                .addStoredField("DefaultMessage.url").addAggregation(aggJava).setSize(1).execute().actionGet();
        long javaCount = srJava.getHits().getTotalHits();
        long java888Count = ((Filter) srJava.getAggregations().get("filter")).getDocCount();

        QueryBuilder querySniffer = QueryBuilders.boolQuery()
                .filter(QueryBuilders.termQuery("transactionId", transactionId));
        SearchResponse srSniffer = client.prepareSearch("apm2.0-" + masterIp + "-snifferdata-1").setQuery(querySniffer)
                .setSize(0).execute().actionGet();
        long snifferCount = srSniffer.getHits().getTotalHits();

        QueryBuilder queryHttp = QueryBuilders.boolQuery()
                .filter(QueryBuilders.termQuery("headers.apm-transactionid", transactionId));
        SearchResponse srHttp = client.prepareSearch("apm2.0-" + masterIp + "-httpdata-1").setQuery(queryHttp)
                .setSize(0).execute().actionGet();
        long httpCount = srHttp.getHits().getTotalHits();

        StringBuffer stringBuffer = new StringBuffer();
        if (javaCount - java888Count != snifferCount || snifferCount != httpCount) {
            correctness = false;
        }

        String url = "";
        if (javaCount == 1 && snifferCount == 0 && httpCount == 0) {
            url = (String) srJava.getHits().getAt(0).getFields().get("DefaultMessage.url").getValue();
        }

        if (!correctness) {
            stringBuffer.append("[Analyzer-After][" + masterIp + "][" + transactionId + "][" + correctness + "] ")
                    .append("javaCount: ").append(javaCount).append(", java888Count: ").append(java888Count)
                    .append(", snifferCount: ").append(snifferCount).append(", httpCount: ").append(httpCount)
                    .append(", url:").append(url);

            System.out.println(stringBuffer);
        }

        return correctness;
    }

    public static void main(String[] args) {
        // 120.26.106.94 hld
        // 122.224.226.98 股权
        // shuiyou_online 税友
        // 120.26.244.78 浙农金服

        DataCorrectnessCheck check = new DataCorrectnessCheck("shuiyou_online");
        Set<String> transactionIds = check.getAllTransactionId(new DateTime(2017, 3, 14, 0, 0));
        for (String transactionId : transactionIds) {
            check.checkTransactionId(transactionId);
        }
    }
}
